package com.cw.base.jeeyc.common.dispatch.exec;

import java.lang.reflect.Constructor;
import java.util.Map;
import java.util.concurrent.Callable;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

import org.springframework.beans.BeansException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.annotation.Lazy;
import org.springframework.lang.NonNull;

import com.cw.base.common.model.util.AppParamUtil;
import com.cw.base.jeeyc.common.dispatch.pojo.Task;
import com.esotericsoftware.reflectasm.MethodAccess;
import com.google.common.collect.Maps;
import com.google.common.util.concurrent.FutureCallback;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.ListeningScheduledExecutorService;
import com.google.common.util.concurrent.MoreExecutors;
import com.cw.base.jeeyc.common.dispatch.event.MessageEvent;
import com.cw.base.jeeyc.common.dispatch.listening.MessageListening;
import com.cw.base.jeeyc.common.dispatch.message.Message;
import com.cw.base.jeeyc.common.dispatch.queue.Queue;
import com.cw.base.jeeyc.common.dispatch.task.facade.pojo.callback.CallBackResponse;
import com.cw.base.jeeyc.common.dispatch.utils.MapDbUtil;

import lombok.extern.slf4j.Slf4j;

/**
 * 将Task给与异步线程池去执行<br>
 * Created on 15/11/10.
 *
 * @author zhangcheng
 */
@Slf4j
public class AsyncExecMessageListening extends MessageListening<MessageEvent> implements ApplicationContextAware {

    private static ApplicationContext applicationContext;
    private final ListeningScheduledExecutorService listeningScheduledExecutorService;
    /** 记录次数 * */
    private Map<String, AtomicInteger> indexMap = Maps.newConcurrentMap();

    @Autowired
    @Qualifier("taskQueue")
    @Lazy
    private Queue queue;

    public AsyncExecMessageListening() {
        ScheduledExecutorService scheduler = new ScheduledThreadPoolExecutor(400, r -> {
            Thread thread = new Thread(r);
            thread.setName("dispatch-task-" + thread.getName());
            return thread;
        });
        listeningScheduledExecutorService = MoreExecutors.listeningDecorator(scheduler);
    }

    @Override
    public void listen(MessageEvent event) {
        Message<Task> message = event.getMessage();
        final Task task = message.getMessage();
        if (task != null) {
            if (null == indexMap.get(task.getUuid())) {
                indexMap.put(task.getUuid(), new AtomicInteger(0));
            }

            long delay = task.getDelayExecTime() + task.getApartExecTime() * indexMap.get(task.getUuid())
                .intValue();
            log.debug("{}下一次补偿间隔时间{}s", task.getUuid(), delay / 1000);
            ListenableFuture explosion = listeningScheduledExecutorService.schedule(new ScheduledCallable(task), delay,
                TimeUnit.MILLISECONDS);

            Futures.addCallback(explosion, new FutureCallback<Object>() {
                @Override
                public void onSuccess(Object result) {
                    AppParamUtil.setRequest(task.getBaseReqDto()
                        .copy());
                    if (null != task.getBaseUserReqDto()) {
                        AppParamUtil.setUserRequest(task.getBaseUserReqDto()
                            .copy());
                    }
                    try {
                        if (task.getLoopCondition()
                            .condition(result)) {
                            log.info("{} 的{}/{}次补偿判定失败，继续补偿", task.getUuid(), indexMap.get(task.getUuid())
                                .intValue() + 1, task.getExecNum());
                            // 业务校验需要继续循环
                            continueLoop(task, result);
                        } else {
                            log.info("{} 的{}/{}次补偿判定成功，开始回调正常处理", task.getUuid(), indexMap.get(task.getUuid())
                                .intValue() + 1, task.getExecNum());
                            // 业务校验停止循环
                            if (task.getTaskCallBackClass() != null) {
                                callBack(task.getTaskCallBackClass(), task.getTaskCallBackMethod(), result, task);
                            }
                            MapDbUtil.deleteData(task.getUuid());
                        }
                    } finally {
                        AppParamUtil.removeRequest();
                        AppParamUtil.removeUserRequest();
                    }
                }

                @Override
                public void onFailure(@NonNull Throwable t) {
                    log.error("{} 的{}/{}次补偿发生异常", task.getUuid(), indexMap.get(task.getUuid())
                        .intValue() + 1, task.getExecNum());
                    AppParamUtil.setRequest(task.getBaseReqDto()
                        .copy());
                    if (null != task.getBaseUserReqDto()) {
                        AppParamUtil.setUserRequest(task.getBaseUserReqDto()
                            .copy());
                    }

                    try {
                        if (task.isErrorLoop()) {
                            log.info("{} 的{}/{}次补偿失败，继续补偿", task.getUuid(), indexMap.get(task.getUuid())
                                .intValue() + 1, task.getExecNum());
                            // 异常需要循环
                            continueLoop(task, t);
                        } else {
                            log.info("{} 的{}/{}次补偿失败，开始回调异常处理", task.getUuid(), indexMap.get(task.getUuid())
                                .intValue() + 1, task.getExecNum());
                            // 异常直接回调
                            if (task.getErrorTaskCallBackClass() != null) {
                                callBack(task.getErrorTaskCallBackClass(), task.getErrorTaskCallBackMethod(), t, task);
                            }
                            MapDbUtil.deleteData(task.getUuid());
                        }
                    } finally {
                        AppParamUtil.removeRequest();
                        AppParamUtil.removeUserRequest();
                    }
                }
            }, MoreExecutors.directExecutor());
        } else {
            log.info("task is null!");
        }
    }

    /**
     * 循环执行任务
     *
     * @param task 补偿任务
     */
    private void continueLoop(final Task task, Object result) {
        if (task.getExecNum() - 1 > indexMap.get(task.getUuid())
            .intValue()) {
            // 未超过执行次数，继续执行
            queue.put(task);
            indexMap.put(task.getUuid(), new AtomicInteger(indexMap.get(task.getUuid())
                .addAndGet(1)));
        } else {
            log.info("{} 的{}/{}次补偿次数用尽，开始回调异常处理", task.getUuid(), indexMap.get(task.getUuid())
                .intValue() + 1, task.getExecNum());
            // 超过执行次数，执行异常回调
            if (task.getErrorTaskCallBackClass() != null) {
                callBack(task.getErrorTaskCallBackClass(), task.getErrorTaskCallBackMethod(), result, task);
            }
            MapDbUtil.deleteData(task.getUuid());
        }
    }

    /**
     * 异步或者正常时进行回调
     *
     * @param callBackClass 回调类全路径或者bean名称
     * @param callBackMethod 回调方法
     * @param result 运行结果，如果发生异常为 Throwable
     * @param task 业务侧发送的任务
     */
    public void callBack(String callBackClass, String callBackMethod, Object result, Task task) {

        if (callBackClass == null || "".equals(callBackClass)) {
            return;
        }
        Object targetObject = null;
        Class clazz = null;
        if (callBackClass.indexOf(".") != -1) {
            try {
                clazz = this.getClass()
                    .getClassLoader()
                    .loadClass(callBackClass);
                if (task.getCallBackConstructorParameters() == null
                    || task.getCallBackConstructorParameters().length == 0) {
                    targetObject = clazz.getDeclaredConstructor().newInstance();
                } else {
                    Class[] classes = new Class[task.getCallBackConstructorParameters().length];
                    for (int i = 0; i < task.getCallBackConstructorParameters().length; i++) {
                        classes[i] = task.getCallBackConstructorParameters()[i].getClass();
                    }
                    Constructor constructor = clazz.getConstructor(classes);
                    targetObject = constructor.newInstance(task.getCallBackConstructorParameters());
                }
            } catch (Exception ex) {
                ex.printStackTrace();
                throw new RuntimeException("无法实例化对象:" + task.getTargetClass());
            }
        } else {
            targetObject = applicationContext.getBean(callBackClass);
            clazz = targetObject.getClass();
        }

        MethodAccess methodAccess = MethodAccess.get(clazz);
        CallBackResponse callBackResponse = CallBackResponse.builder()
            .businessParam(task.getBusinessParam())
            .baseUserReqDto(task.getBaseUserReqDto())
            .request(task.getBaseReqDto())
            .result(result)
            .build();
        methodAccess.invoke(targetObject, callBackMethod, callBackResponse);
    }

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        this.applicationContext = applicationContext;
    }

    /**
     * 线程
     */
    private class ScheduledCallable implements Callable<Object> {
        private Task task;

        private ScheduledCallable(Task task) {
            this.task = task;
        }

        @Override
        public Object call() {
            // 设置公共上下文
            AppParamUtil.setRequest(task.getBaseReqDto()
                .copy());
            if (null != task.getBaseUserReqDto()) {
                AppParamUtil.setUserRequest(task.getBaseUserReqDto()
                    .copy());
            }
            try {
                return execute();
            } finally {
                AppParamUtil.removeRequest();
                AppParamUtil.removeUserRequest();
            }
        }

        /**
         * 执行业务方法
         *
         * @return 业务返回结果
         */
        private Object execute() {
            Object result;
            if (task.getNormalTask() != null) {
                result = task.getNormalTask()
                    .call();
            } else {
                Object targetObject = null;
                Class clazz = null;
                if (task.getTargetClass() != null && task.getTargetClass()
                    .indexOf(".") != -1) {
                    try {
                        clazz = this.getClass()
                            .getClassLoader()
                            .loadClass(task.getTargetClass());
                        if (task.getConstructorParameters() == null || task.getConstructorParameters().length == 0) {
                            targetObject = clazz.getDeclaredConstructor().newInstance();
                        } else {
                            Class[] classes = new Class[task.getConstructorParameters().length];
                            for (int i = 0; i < task.getConstructorParameters().length; i++) {
                                classes[i] = task.getConstructorParameters()[i].getClass();
                            }
                            Constructor constructor = clazz.getConstructor(classes);
                            targetObject = constructor.newInstance(task.getConstructorParameters());
                        }
                    } catch (Exception ex) {
                        ex.printStackTrace();
                        throw new RuntimeException("无法实例化对象:" + task.getTargetClass());
                    }
                } else {
                    targetObject = applicationContext.getBean(task.getTargetClass());
                    clazz = targetObject.getClass();
                }

                MethodAccess methodAccess = MethodAccess.get(clazz);
                result = methodAccess.invoke(targetObject, task.getTargetMethod(), task.getParameters());
                task.setResultObject(result);
            }
            return result;
        }
    }
}
